package cn.edu.thu.laud.jdbc;


import java.nio.ByteBuffer;
import java.sql.Array;
import java.sql.Blob;
import java.sql.CallableStatement;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.NClob;
import java.sql.PreparedStatement;
import java.sql.SQLClientInfoException;
import java.sql.SQLException;
import java.sql.SQLNonTransientConnectionException;
import java.sql.SQLWarning;
import java.sql.SQLXML;
import java.sql.Savepoint;
import java.sql.Statement;
import java.sql.Struct;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.antlr.runtime.ANTLRStringStream;
import org.antlr.runtime.CharStream;
import org.antlr.runtime.CommonTokenStream;
import org.antlr.runtime.TokenStream;
import org.apache.cassandra.cql.jdbc.LaUDCassandraConnection;
import org.apache.cassandra.cql.jdbc.LaUDCassandraUtils;
import org.apache.cassandra.cql.jdbc.LaUDCassandraUtils.UriProperties;
import org.apache.cassandra.thrift.AuthenticationException;
import org.apache.cassandra.thrift.AuthenticationRequest;
import org.apache.cassandra.thrift.AuthorizationException;
import org.apache.cassandra.thrift.Compression;
import org.apache.cassandra.thrift.CqlResult;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SchemaDisagreementException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.hadoop.hive.jdbc.LaUDHiveConnection;
import org.apache.hadoop.hive.service.HiveServerException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.edu.thu.laud.lasql.parser.LaSQLEasyLexer;
import cn.edu.thu.laud.lasql.parser.LaSQLEasyParser;
import cn.edu.thu.laud.lasql.parser.LaSQLFunctionParser;
import cn.edu.thu.laud.server.thrift.LaUDResult;
import cn.edu.thu.laud.server.thrift.LaUDServerException;
import cn.edu.thu.laud.thrift.LaUDClient;


import com.google.common.base.Charsets;

public class LaUDConnection implements Connection {
	private LaUDClient client;
	private TTransport transport;
	protected String username = null;

	public String host;
	public int port;
	public int dataPort;

	private boolean support_function=false; 


	//protected String url = null;
	@SuppressWarnings("unused")
	private static final Logger logger = LoggerFactory.getLogger(LaUDConnection.class);  

	private LaUDCassandraConnection cassandraConnection;

	private LaUDHiveConnection hiveConnection;

	Set<LaUDStatement> statements=new HashSet<LaUDStatement>();

	public boolean includeHive=false;
	/**
	 * 
	 * @param hiveUrl 
	 * @param props 
	 * @throws SQLException
	 */
	public LaUDConnection(UriProperties uriProperties) throws  SQLException {
		Properties props=uriProperties.properties;
		try
		{
			host = props.getProperty(LaUDCassandraUtils.TAG_SERVER_NAME);
			port = Integer.parseInt(props.getProperty(LaUDCassandraUtils.TAG_PORT_NUMBER));
			String keyspace = props.getProperty(LaUDCassandraUtils.TAG_DATABASE_NAME);
			username = props.getProperty(LaUDCassandraUtils.TAG_USER);
			String password = props.getProperty(LaUDCassandraUtils.TAG_PASSWORD);

			TSocket socket = new TSocket(host, port,0);
			transport = new TFramedTransport(socket);
			TProtocol protocol = new TBinaryProtocol(transport);
			setClient(new LaUDClient(protocol));
			socket.open();
			//login
			if (username != null)
            {
                Map<String, String> credentials = new HashMap<String, String>();
                credentials.put("username", username);
                if (password != null) credentials.put("password", password);
                AuthenticationRequest areq = new AuthenticationRequest(credentials);
                client.getCassandraClient().login(areq);

            }
			client.setSessionId();
			TTransport noCloseTransport=new LaUDClient.LaUDNullCloseTTransport(transport);
			//cassandraConnection to init the ColumnDecode.
			cassandraConnection=new LaUDCassandraConnection(props, getClient().getCassandraClient(), noCloseTransport);
			if(uriProperties.uri.getQuery()!=null&&uriProperties.uri.getQuery().contains("hive=true")){
				//hiveConnection to init config.
				hiveConnection=new LaUDHiveConnection(getClient().getHiveClient(),noCloseTransport);
				includeHive=true;
			}
			dataPort=Integer.parseInt( (String) (uriProperties.properties.get("data_port")==null?((port+1)+""):uriProperties.properties.get("data_port")));
			this.support_function=uriProperties.properties.get("support_function")==null?false:true;

		}
		catch (TException e)
		{
			throw new SQLNonTransientConnectionException(e);
		} catch (LaUDServerException e) {
			throw new SQLNonTransientConnectionException(e);
		} catch (AuthenticationException e) {
			throw new SQLNonTransientConnectionException(e);
		} catch (AuthorizationException e) {
			throw new SQLNonTransientConnectionException(e);
		}
	}

	@Override
	public boolean isWrapperFor(Class<?> iface) throws SQLException {
		return (iface.isAssignableFrom(LaUDCassandraConnection.class)||iface.isAssignableFrom(LaUDHiveConnection.class));
	}

	@Override
	public <T> T unwrap(Class<T> iface) throws SQLException {
		if(iface.isAssignableFrom(LaUDCassandraConnection.class)){
			return iface.cast(LaUDCassandraConnection.class);
		}else if(iface.isAssignableFrom(LaUDHiveConnection.class)){
			return iface.cast(LaUDHiveConnection.class);
		}else{
			return cassandraConnection.unwrap(iface);
		}
	}

	@Override
	public void clearWarnings() throws SQLException {
		cassandraConnection.clearWarnings();
		hiveConnection.clearWarnings();

	}

	@Override
	public void close() throws SQLException {
		if(!isClosed()){
			if(includeHive)
				this.hiveConnection.close();
			this.cassandraConnection.close();
			for (Statement statement : statements)
				statement.close();
			statements.clear();
			if(this.transport.isOpen()){
				this.transport.close();
			}
		}
	}
	/**
	 * neighter hive nor cassandra support it. 
	 */
	@Override
	public void commit() throws SQLException {
		String exception="";
		try{
			cassandraConnection.commit();
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			hiveConnection.commit();
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(!exception.equals("")){
			throw new SQLException(exception);
		}
	}

	@Override
	public Array createArrayOf(String typeName, Object[] elements)
			throws SQLException {
		String exception="";
		try{
			cassandraConnection.createArrayOf(typeName, elements);
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			hiveConnection.createArrayOf(typeName, elements);
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(!exception.equals("")){
			throw new SQLException(exception);
		}
		//TODO return the right result.
		return null;
	}

	@Override
	public Blob createBlob() throws SQLException {
		String exception="";
		try{
			cassandraConnection.createBlob();
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			hiveConnection.createBlob();
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(!exception.equals("")){
			throw new SQLException(exception);
		}
		//TODO return the right result.
		return null;
	}

	@Override
	public Clob createClob() throws SQLException {
		String exception="";
		try{
			cassandraConnection.createClob();
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			hiveConnection.createClob();
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(!exception.equals("")){
			throw new SQLException(exception);
		}
		//TODO return the right result.
		return null;
	}

	@Override
	public NClob createNClob() throws SQLException {
		String exception="";
		try{
			cassandraConnection.createNClob();
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			hiveConnection.createNClob();
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(!exception.equals("")){
			throw new SQLException(exception);
		}
		//TODO return the right result.
		return null;
	}

	@Override
	public SQLXML createSQLXML() throws SQLException {
		String exception="";
		try{
			cassandraConnection.createSQLXML();
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			hiveConnection.createSQLXML();
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(!exception.equals("")){
			throw new SQLException(exception);
		}
		//TODO return the right result.
		return null;
	}

	@Override
	public Statement createStatement() throws SQLException {
		LaUDStatement statement=new LaUDStatement(this, cassandraConnection.createStatement(),includeHive?hiveConnection.createStatement():null);
		checkNotClosed();
		statements.add(statement);
		return statement;
	}

	private void checkNotClosed() throws SQLException {
		if(isClosed()){
			throw new SQLException("connection is closed.");
		}
	}

	@Override
	public Statement createStatement(int resultSetType, int resultSetConcurrency)
			throws SQLException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Statement createStatement(int resultSetType,
			int resultSetConcurrency, int resultSetHoldability)
					throws SQLException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Struct createStruct(String typeName, Object[] attributes)
			throws SQLException {
		String exception="";
		try{
			cassandraConnection.createStruct(typeName, attributes);
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			hiveConnection.createStruct(typeName, attributes);
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(!exception.equals("")){
			throw new SQLException(exception);
		}
		//TODO return the right result.
		return null;
	}

	@Override
	public boolean getAutoCommit() throws SQLException {
		String exception="";
		try{
			cassandraConnection.getAutoCommit();
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			hiveConnection.getAutoCommit();
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(!exception.equals("")){
			throw new SQLException(exception);
		}
		return cassandraConnection.getAutoCommit()&&hiveConnection.getAutoCommit();
	}

	@Override
	public String getCatalog() throws SQLException {
		String exception="";
		try{
			cassandraConnection.getCatalog();
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			hiveConnection.getCatalog();
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(!exception.equals("")){
			throw new SQLException(exception);
		}
		throw new SQLException("not support now.");
	}

	@Override
	public Properties getClientInfo() throws SQLException {
		Properties properties=new Properties();
		Properties temp=null;
		String exception="";

		try{
			properties=cassandraConnection.getClientInfo();
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			temp=hiveConnection.getClientInfo();
			for(Object key:temp.keySet()){
				properties.put(key, temp.get(key));
			}
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(	properties.size()==0&&!exception.equals("")){
			throw new SQLException(exception);
		}
		return properties;
	}
	/**
	 * return both cassandra and hive .
	 * <br>type:${cassandra}<b>content</b>${hive}<B>content</b>
	 */
	@Override
	public String getClientInfo(String name) throws SQLException {
		StringBuilder infoBuilder=new StringBuilder("${cassandra}");
		try{
			infoBuilder.append(cassandraConnection.getClientInfo(name));
		}catch (SQLException e) {	
		}
		infoBuilder.append("${hive}");
		try{
			infoBuilder.append(hiveConnection.getClientInfo(name));
		}catch (SQLException e) {	

		}
		if(infoBuilder.length()==19)
			return null;
		else
			return infoBuilder.toString();
	}

	@Override
	public int getHoldability() throws SQLException {
		String exception="";
		int c=0;
		boolean supportc=false;
		int h=0;
		boolean supporth=false;
		try{
			c=cassandraConnection.getHoldability();
			supportc=true;
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			h=hiveConnection.getHoldability();
			supporth=true;
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(exception.equals("")){
			if(c==h)
				return c;
			else {
				throw new SQLException("cassandra result is "+c+" but hive result is "+h);
			}
		}			
		else{
			if(supportc){
				throw new SQLException("cassandra rsult is "+c+"but hive "+exception);
			}else if(supporth){
				throw new SQLException("hive rsult is "+h+"but cassandra "+exception);
			}else {
				throw new SQLException(exception);
			}
		}
	}

	@Override
	public DatabaseMetaData getMetaData() throws SQLException {
		throw new SQLException("try to use getHiveMetaData or getCassandraMetaData instead.");
	}
	public DatabaseMetaData getCassandraMetaData() throws SQLException {
		return cassandraConnection.getMetaData();
	}
	public DatabaseMetaData getHiveMetaData() throws SQLException {
		return hiveConnection.getMetaData();
	}
	@Override
	public int getTransactionIsolation() throws SQLException {
		String exception="";
		int c=0;
		boolean supportc=false;
		int h=0;
		boolean supporth=false;
		try{
			c= cassandraConnection.getTransactionIsolation();
			supportc=true;
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			h=hiveConnection.getTransactionIsolation();
			supporth=true;
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(exception.equals("")){
			if(c==h)
				return c;
			else {
				throw new SQLException("cassandra result is "+c+" but hive result is "+h);
			}
		}			
		else{
			if(supportc){
				throw new SQLException("cassandra rsult is "+c+"but hive "+exception);
			}else if(supporth){
				throw new SQLException("hive rsult is "+h+"but cassandra "+exception);
			}else {
				throw new SQLException(exception);
			}
		}

	}

	@Override
	public Map<String, Class<?>> getTypeMap() throws SQLException {
		String exception="";
		Map<String, Class<?>> map=null;
		try{
			map=cassandraConnection.getTypeMap();
		}catch (SQLException e) {	
			exception=e.getMessage()+"\tand\t";
		}
		try{
			if(map==null)
				map=hiveConnection.getTypeMap();
			else
				map.putAll(hiveConnection.getTypeMap());
		}catch(SQLException e){
			exception+=e.getMessage();
		}
		if(!exception.equals("")){
			throw new SQLException(exception);
		}
		return null;
	}

	@Override
	public SQLWarning getWarnings() throws SQLException {
		SQLWarning warning=null;
		try{
			warning=(cassandraConnection.getWarnings());
		}catch (SQLException e) {
		}
		if(warning==null){
			warning=hiveConnection.getWarnings();
		}else
			warning.setNextWarning(hiveConnection.getWarnings());
		return warning;
	}

	@Override
	public boolean isClosed() throws SQLException {
		return !transport.isOpen();
	}

	@Override
	public boolean isReadOnly() throws SQLException {
		return cassandraConnection.isReadOnly()&&hiveConnection.isReadOnly();
	}

	@Override
	public boolean isValid(int timeout) throws SQLException {
		return cassandraConnection.isValid(timeout)&&hiveConnection.isValid(timeout);

	}

	@Override
	public String nativeSQL(String sql) throws SQLException {
		String lasql=cassandraConnection.nativeSQL(sql);
		//TODO we can do some native funtion here.
		return lasql;
	}

	@Override
	public CallableStatement prepareCall(String sql) throws SQLException {
		throw new SQLException("not support now.");
	}

	@Override
	public CallableStatement prepareCall(String sql, int resultSetType,
			int resultSetConcurrency) throws SQLException {
		throw new SQLException("not support now.");	}

	@Override
	public CallableStatement prepareCall(String sql, int resultSetType,
			int resultSetConcurrency, int resultSetHoldability)
					throws SQLException {
		throw new SQLException("not support now.");
	}

	@Override
	public PreparedStatement prepareStatement(String sql) throws SQLException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys)
			throws SQLException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public PreparedStatement prepareStatement(String sql, int[] columnIndexes)
			throws SQLException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public PreparedStatement prepareStatement(String sql, String[] columnNames)
			throws SQLException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public PreparedStatement prepareStatement(String sql, int resultSetType,
			int resultSetConcurrency) throws SQLException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public PreparedStatement prepareStatement(String sql, int resultSetType,
			int resultSetConcurrency, int resultSetHoldability)
					throws SQLException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public void releaseSavepoint(Savepoint savepoint) throws SQLException {
		try{
			cassandraConnection.releaseSavepoint(savepoint);
		}catch(SQLException e){

		}
		try{
			hiveConnection.releaseSavepoint(savepoint);
		}catch(SQLException e){

		}
		throw new SQLException("not support now.");
	}

	@Override
	public void rollback() throws SQLException {
		try{
			cassandraConnection.rollback();
		}catch(SQLException e){

		}
		try{
			hiveConnection.rollback();
		}catch(SQLException e){

		}
		throw new SQLException("not support now.");
	}

	@Override
	public void rollback(Savepoint savepoint) throws SQLException {
		try{
			cassandraConnection.rollback(savepoint);
		}catch(SQLException e){

		}
		try{
			hiveConnection.rollback(savepoint);
		}catch(SQLException e){

		}
		throw new SQLException("not support now.");
	}

	@Override
	public void setAutoCommit(boolean autoCommit) throws SQLException {
		try{
			cassandraConnection.setAutoCommit(autoCommit);
		}catch(SQLException e){

		}
		try{
			hiveConnection.setAutoCommit(autoCommit);
		}catch(SQLException e){

		}
		throw new SQLException("not support now.");
	}

	@Override
	public void setCatalog(String catalog) throws SQLException {
		try{
			cassandraConnection.setCatalog(catalog);
		}catch(SQLException e){

		}
		try{
			hiveConnection.setCatalog(catalog);
		}catch(SQLException e){

		}
		throw new SQLException("not support now.");
	}

	@Override
	public void setClientInfo(Properties properties)
			throws SQLClientInfoException {
		try{
			cassandraConnection.setClientInfo(properties);
		}catch(SQLException e){

		}
		try{
			hiveConnection.setClientInfo(properties);
		}catch(SQLException e){

		}
		throw new SQLClientInfoException();
	}

	@Override
	public void setClientInfo(String name, String value)
			throws SQLClientInfoException {
		try{
			cassandraConnection.setClientInfo(name,value);
		}catch(SQLException e){

		}
		try{
			hiveConnection.setClientInfo(name,value);
		}catch(SQLException e){

		}
		throw new SQLClientInfoException();

	}

	@Override
	public void setHoldability(int holdability) throws SQLException {
		try{
			cassandraConnection.setHoldability(holdability);
		}catch(SQLException e){

		}
		try{
			hiveConnection.setHoldability(holdability);
		}catch(SQLException e){

		}
		throw new SQLException("not support now");
	}

	@Override
	public void setReadOnly(boolean readOnly) throws SQLException {
		try{
			cassandraConnection.setReadOnly(readOnly);
		}catch(SQLException e){

		}
		try{
			hiveConnection.setReadOnly(readOnly);
		}catch(SQLException e){

		}
		throw new SQLException("not support now");

	}

	@Override
	public Savepoint setSavepoint() throws SQLException {
		try{
			cassandraConnection.setSavepoint();
		}catch(SQLException e){

		}
		try{
			hiveConnection.setSavepoint();
		}catch(SQLException e){

		}
		throw new SQLException("not support now");
	}

	@Override
	public Savepoint setSavepoint(String name) throws SQLException {
		try{
			cassandraConnection.setSavepoint(name);
		}catch(SQLException e){

		}
		try{
			hiveConnection.setSavepoint(name);
		}catch(SQLException e){

		}
		throw new SQLException("not support now");
	}

	@Override
	public void setTransactionIsolation(int level) throws SQLException {
		try{
			cassandraConnection.setTransactionIsolation(level);
		}catch(SQLException e){

		}
		try{
			hiveConnection.setTransactionIsolation(level);
		}catch(SQLException e){

		}
		throw new SQLException("not support now");
	}

	@Override
	public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
		try{
			cassandraConnection.setTypeMap(map);
		}catch(SQLException e){

		}
		try{
			hiveConnection.setTypeMap(map);
		}catch(SQLException e){

		}

		throw new SQLException("not support now");
	}

	LaUDResult execute(String lasql,Compression compression) throws LaUDServerException, TException, InvalidRequestException, UnavailableException, TimedOutException, SchemaDisagreementException, HiveServerException{
		ByteBuffer query=null;
		switch (compression) {
		case GZIP:
			query=LaUDCassandraUtils.compressQuery(lasql, compression);
			break;
		case NONE:
			query=ByteBuffer.wrap(lasql.getBytes(Charsets.UTF_8));
			break;
		default:
			throw new LaUDServerException().setMessage("unsupport the compression type.");
		}
		if(support_function){
			CharStream stream = new ANTLRStringStream(lasql);
			LaSQLEasyLexer lexer = new LaSQLEasyLexer(stream);
			TokenStream tokenStream = new CommonTokenStream(lexer);
			LaSQLEasyParser parser = new LaSQLEasyParser(tokenStream);
			List<LaSQLFunctionParser> functions=new ArrayList<LaSQLFunctionParser>();
			List<ByteBuffer> functionValues=new ArrayList<ByteBuffer>(functions.size());
			
			try{
				parser.query(functions);
			}catch (Exception e) {
			}finally{
				for(LaSQLFunctionParser function:functions){
					try {
						functionValues.add(function.getResult());
					} catch (Exception e) {
						throw new InvalidRequestException().setWhy(e.getMessage());
					}
				}
			}
			return this.getClient().execute_with_parameters(query,compression,functionValues);
		}
		return this.getClient().execute(query,compression);
	}

	public boolean removeStatement(LaUDStatement laUDStatement) {
		return statements.remove(laUDStatement);
	}

	void setCassandraResult(CqlResult fakeCqlResult){
		((LaUDClient.LaUDCassandraClient)this.getClient().getCassandraClient()).setFakeCqlResult(fakeCqlResult);
	}

	public LaUDClient getClient() {
		return client;
	}

	public void setClient(LaUDClient client) {
		this.client = client;
	}

}
